package com.globalwave.common.task;

import java.sql.Connection;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.globalwave.common.ArrayPageList;
import com.globalwave.common.entity.SyncBatch;
import com.globalwave.common.scheduling.RunnableTask;
import com.globalwave.common.service.SyncBatchService;
import com.globalwave.common.service.SyncBatchServiceImpl;
import com.globalwave.datasync.SyncContext;
import com.globalwave.datasync.engine.OneByOneSync;

public class SyncBatchTask extends RunnableTask {

	protected Log log = LogFactory.getLog(this.getClass()) ;
	
	private SyncBatchService syncBatchService ;
	
	protected int threadId = -1 ; // 未支持多线程
	protected String sourceDbCode = null ;// 确定数据源
	protected String batchDbCode = null ;// XXbatch表所在数据库代码
	
	public void execute() throws Exception {

		sourceDbCode = (String)super.getContext().get("sourceDbCode") ;
		batchDbCode = (String)super.getContext().get("batchDbCode") ;
		Connection batchConnection = connectionProvider.getConnection(batchDbCode) ;
		syncBatchService = new SyncBatchServiceImpl(batchConnection) ;

    	batchConnection.setAutoCommit(true) ;
    	try {
	        if (this.threadId == -1 || this.threadId == 4) {// 要么开-1线程，要么开0~9线程(一个线程做清理就OK了，一般4号线程负载小)
	            syncBatchService.resetToSyncable(null) ;
	        }
	        
	        Long startId = 0L ;
	        
	        ArrayPageList<SyncBatch> batches = syncBatchService.querySyncableBatch(startId, threadId);
	
	        if (batches.size() == 0) {
	        	return ;
	        }
			SyncContext context = new SyncContext() ;
			context.put("sourceDbCode", sourceDbCode) ;
	    		
			final OneByOneSync sync = new OneByOneSync(this.connectionProvider);
				
	    	while(batches.size() > 0){
	
	    		SyncBatch batch = (SyncBatch) batches.remove(0) ;
	   
	    		if (batches.size() <= 0) {
		    	    if (!super.isCanceled()) {
		    	    	startId = batch.getSync_batch_id() + 1 ;
		    		    batches = syncBatchService.querySyncableBatch(startId, threadId);
		    	    }
		        }
		    
				try {
					context.put("batch", batch) ;
					
		    		sync.execute(context) ;
				    
			    	syncBatchService.updateSuccess(batch) ;
			    	
				} catch (Exception e) {
	
				    log.error("batch[id=" + batch.getSync_batch_id() + "] sync fail!",e) ;// 后台打印出差信息
				    
			    	syncBatchService.updateFail(batch, e) ;
				} 
				
			}
    	} finally {
    	    batchConnection.setAutoCommit(false) ;
    	}

	}

}
